package com.example.finalwork2.component;

import com.example.finalwork2.entity.VehicleSignal;
import com.example.finalwork2.service.VehicleSignalService;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;

@Component
@RequiredArgsConstructor
public class CacheSyncScheduler {

    private final JedisPool jedisPool;
    private final ObjectMapper objectMapper;
    private final VehicleSignalService vehicleSignalService; // 注入Service
    private final RocketMQProducerService rocketMQProducerService;

    @Scheduled(fixedRate = 60_00)
    public void syncSignalsToDbAndMq() {
        try (Jedis jedis = jedisPool.getResource()) {
            Set<String> keys = jedis.keys("vehicle:signal:*");

            //当前时间戳
            long now = LocalDateTime.now().toEpochSecond(ZoneOffset.UTC);
            //long now = Instant.now().getEpochSecond();
            //long now = Instant.now().getEpochSecond();
            System.out.println(now);

            for (String key : keys) {
                String vehicleId = key.substring("vehicle:signal:".length());
                String cursorKey = "last_processed_time:" + vehicleId;

                // 获取上次处理时间戳
                String lastProcessedStr = jedis.get(cursorKey);
                long lastProcessed = lastProcessedStr != null ? Long.parseLong(lastProcessedStr) : 0;

                System.out.println("处理车辆：" + vehicleId);
                System.out.println("上次处理时间：" + lastProcessed);

                // 查询未处理数据
               List<String> newJsonList = jedis.zrangeByScore(key, "(" + lastProcessed, String.valueOf(now));
                //Set<String> values = jedis.zrangeByScore;
                System.out.println("处理数据：" + newJsonList.size());

                if (newJsonList.isEmpty()) continue;

                List<VehicleSignal> signalList = new ArrayList<>();
                for (String json : newJsonList) {
                    try {
                        VehicleSignal signal = objectMapper.readValue(json, VehicleSignal.class);
                        signalList.add(signal);
                        rocketMQProducerService.sendMessage("test-topic", json);
                    } catch (Exception e) {
                        e.printStackTrace(); // JSON反序列化失败跳过
                    }
                }

                // 存入数据库
                if (!signalList.isEmpty()) {
                    vehicleSignalService.saveBatch(signalList);
                }


                jedis.set(cursorKey, String.valueOf(now));

                //
                long sevenDaysAgo = now - 7 * 24 * 3600;
                jedis.zremrangeByScore(key, 0, sevenDaysAgo);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}

